package main

import (
	"fmt"
	"path"
	"sync"
	"time"

	"../auth"
	"../prometheus"
	"github.com/gin-gonic/gin"
	"github.com/kpango/glg"
	crono "github.com/utahta/go-cronowriter"
)

type Controller struct {
	basepath  string
	logname   string
	logger    *crono.CronoWriter
	auth      *auth.Authenticator
	kafka     KafkaHandler
	req_count map[string]int64
	delay     map[string]int64
	ms        *prometheus.MatrixSet
	mu        *sync.RWMutex
}

func NewController(basepath, logname string) (Controller, error) {
	c := Controller{
		logname:	logname,
		basepath:	basepath,
		req_count:	make(map[string]int64),
		delay:		make(map[string]int64),
		mu:			new(sync.RWMutex),
	}
	c.ms = prometheus.NewMatrixSet("crmlog-store")

	logpath := path.Join(c.basepath, c.logname)
	c.logger = crono.MustNew(logpath+".%Y%m%d", crono.WithInit())
	glg.Get().SetMode(glg.WRITER).AddWriter(c.logger)

	auth, err := auth.NewAuthenticator(*AuthPath)
	if err != nil {
		glg.Errorf(err.Error())
		return c, err
	}
	c.auth = auth

	kafka, err := NewKafkaHandler()
	if err != nil {
		glg.Errorf(err.Error())
		return c, err
	} else {
		glg.Infof("storeService.NewController: Created Kafka producer successfully: Url:%s, Topic:%s", *KUrl, *KTopic)
	}
	c.kafka = kafka
	return c, nil
}

func (c *Controller) Close() {
	c.auth.Close()
	if err := c.kafka.Close(); err != nil {
		glg.Errorf("storeService.Close: Failed to close Kafka producer: %v", err)
	}
	if err := c.logger.Close(); err != nil {
		glg.Errorf("storeService.Close: Failed to close logger: %v", err)
	}
}

func (c *Controller) PostHandler(ctx *gin.Context) {
	t := time.Now().UnixNano() / 1000 / 1000
	method := ctx.PostForm("method")
	tablename := ctx.PostForm("tablename")
	appId := ctx.PostForm("appId")
	data := ctx.PostForm("data")
	timestamp := ctx.PostForm("timestamp")
	signature := ctx.PostForm("signature")
	version := ctx.PostForm("version")
	ip := ctx.ClientIP()
	glg.Infof("storeService.PostHandler: Received a request from: %s for service: "+
		"appName:HTableService,appVersion:%s", ip, version)
	if body, err := ctx.GetRawData(); err != nil {
		glg.Infof("storeService.PostHandler: Request: request.toString(): %s", string(body))
	}
	glg.Infof("storeService.PostHandler: data list:%s", data)
	if method != "savebydatas" {
		desc := fmt.Sprintf("unsolved request method: %v", method)
		c.Fail(ctx, desc, t)
		return
	}
	err := c.auth.Auth(appId, timestamp, signature)
	if err == nil {
		glg.Infof("auth.Auth: Authorization successful for %s/%s", appId, ctx.Request.Host)
	} else if err.Error() == "Invalid appId" {
		desc := fmt.Sprintf("auth.Auth: Authorization failed: No password for %s", appId)
		glg.Errorf(desc)
		c.Fail(ctx, desc, t)
		return
	} else {
		desc := fmt.Sprintf("auth.Auth: Authorization failed: Wrong signature, %s", appId)
		glg.Errorf(desc)
		c.Fail(ctx, desc, t)
		return
	}

	tableInfo := fmt.Sprintf("%s/%s", appId, tablename)
	var offset int64
	if offset, err = c.kafka.Produce(tableInfo, data); err != nil {
		c.Fail(ctx, err.Error(), t)
		return
	}
	c.Success(ctx, offset, t)
}

func (c *Controller) Success(ctx *gin.Context, offset, timestamp int64) {
	delta := time.Now().UnixNano()/1000/1000 - timestamp
	c.mu.Lock()
	c.req_count["request_success"]++
	c.req_count["request_total"]++
	m_offset := make(map[string]int64)
    m_offset["offset"] = offset
	c.delay["delay_success"] = c.delay["delay_success"] + delta
	c.delay["delay_total"] = c.delay["delay_total"] + delta
	c.ms.UpdateMatrix(c.req_count, "request", "request_count")
	c.ms.UpdateMatrix(c.delay, "ms", "delay")
	c.ms.UpdateMatrix(m_offset, "", "kafka_offset")
	c.mu.Unlock()
	ctx.JSON(200, gin.H{
		"desc":       fmt.Sprintf("cost %dms", delta),
		"resultCode": "0",
	})
	glg.Infof("storeService.PostHandler: The Request is OK and cost %v ms", delta)
}

func (c *Controller) Fail(ctx *gin.Context, desc string, timestamp int64) {
	delta := time.Now().UnixNano()/1000/1000 - timestamp
	c.mu.Lock()
	c.req_count["request_fail"]++
	c.req_count["request_total"]++
	c.delay["delay_fail"] = c.delay["delay_fail"] + delta
	c.delay["delay_total"] = c.delay["delay_total"] + delta
	c.ms.UpdateMatrix(c.req_count, "request", "request_count")
	c.ms.UpdateMatrix(c.delay, "ms", "delay")
	c.mu.Unlock()
	ctx.JSON(200, gin.H{
		"desc":       desc,
		"resultCode": "4002",
	})
}

